package framework.service.cluster;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.StreamCorruptedException;
import java.util.Random;
import java.util.Map.Entry;

import framework.Message;
import framework.Trait;
import framework.server.Loader;
import framework.server.Mail;
import framework.server.Server;
import framework.server.ServerException;
import framework.service.AbstractService;
import framework.service.client.ClientManager;
import framework.service.client.ClientMessage;
import framework.service.client.ClientRequest;

/**
 * Dispatcher.<br>
 *
 * @author	Orlin Tomov
 * @version	0.2
 *
 */
public class Dispatcher extends AbstractService {

	private ClientRequest conn = null;
	protected String remote;

	/**
	 * @param	container
	 * @param	host
	 * @param	port
	 * @param	clientPort
	 * @throws	IOException
	 */
	public Dispatcher(Trait container, String host, int port, int clientPort) throws IOException {
		super(container, host + ":" + port);
		conn = new ClientRequest(this, host, port);
		handShaking(clientPort);
	}

	/**
	 * @param	container
	 * @param	clientRequest
	 * @throws	IOException
	 */
	public Dispatcher(Trait trait, ClientRequest clientRequest) throws IOException {
		super(trait, clientRequest.getHost() + ":" + clientRequest.getClientPort());
		remote = ((String) clientRequest.lines.get(0)).substring(0, ((String) clientRequest.lines.get(0)).indexOf(":"));
		this.conn = clientRequest;
		handShaking(-1);
		conn.setInfinite();
	}

	@Override
	protected Server getContainer() {
		return (Server) super.getContainer();
	}

	/**
	 * 
	 * @throws ServerException 
	 * @see	framework.Service#init()
	 */
	@Override
	public void init() throws ServerException {
		if (remote == null) {
			//String line;
			if (conn.lines.size() > 0) {
				remote = (String) conn.lines.get(0);
				conn.lines.remove(0);
			} else {
				try {
					remote = readLine(Server.version.length());
					if (readLine(0).length() > 0) {
						log.warn("ERROR: " + remote);
					}
				} catch (IOException e) {
					log.error(e.getMessage(), e);
				}
			}
			//log.debug("hand shaking: " + line);
			if (!remote.equalsIgnoreCase(Server.version)) {
				throw new ServerException("Unknown dispatcher!");
			}
			conn.setInfinite();
		}
	}

	public void handShaking(int clientPort) throws IOException {
		if (clientPort < 0) {
			writeln(Server.version);
		} else {
			writeln(Server.version + ":" + clientPort);
		}
		writeln();
	}

	private static final String HI = "hi:";

	@Override
	public void run() {
		try {
			init();
			ForwardMessage forwardMsg = null;
			int number = 0;
			int count = 10;
			while (getContainer().isRunning() && isRunning()) {
				if (isRunning()) {
					/** when appears a message to forwarding is generating a random number
					 * to be used 10 times */ 
					if (forwardMsg == null) {
						forwardMsg = getContainer().pullMessage(getName());
					}
					if (forwardMsg == null) {
						number = getContainer().queueSize();
					} else {
						if (--count < 0) {
							number = new Random().nextInt();
							count = 10;
						}
					}
					writeln(HI + number);
					String line = readLine(100);
					//log.debug("response: " + line);
					if (line.startsWith(HI)) {
						String size = line.substring(HI.length());
						int respNumb = Integer.parseInt(size);
						if (number > respNumb) {
							//log.info("response: " + line + "-" + queueSize + "?" + queueResponse);
							/** send message */
							Entry<String, Mail> entry = null;
							if (forwardMsg != null) {
								entry = forwardMsg.getEntry();
							} else {
								entry = getContainer().pullMailEntry();
							}
							if (entry != null) {
								String service = entry.getKey();
								writeln("do?");
								String answer;
								/**/do {
									answer = readLine(100);
									//log.debug("answer: " + answer);
									/**/} while (!answer.equalsIgnoreCase("mm?"));
								writeln(service);
								answer = readLine(3);
								Mail mail = entry.getValue();
								ClientMessage message = mail.getMessage();
								if (!answer.equalsIgnoreCase("ok.")) {
									if (forwardMsg != null) {
										ClientManager clientManager = getContainer().getService(ClientManager.class.getName());
										clientManager.nextClient(message, getName());
										forwardMsg = null;
									} else {
										getContainer().sendMail(mail, service);
									}
									continue;
								}
								writeln(mail.getSender());
								writeln(mail.getMessageName());
								write(mail.getMessage());
								//log.debug("send: " + message.getMessage() + " from " + message.getSender() + " to " + service);
								answer = readLine(3);
								//log.debug("answer: " + answer);
								if (!answer.equalsIgnoreCase("ok.")) {
									if (forwardMsg != null) {
										ClientManager clientManager = getContainer().getService(ClientManager.class.getName());
										clientManager.nextClient(message, getName());
										forwardMsg = null;
									} else {
										getContainer().sendMail(mail, service);
									}
									continue;
								}
								forwardMsg = null;
								log.debug(">" + mail.getMessageName() + " from " + mail.getSender() + " to " + service + " :" + mail.getMessage());
							} else {
								try {
									sleep(100L);
								} catch (InterruptedException exception) {
									/**/log.error(exception.getMessage(), exception);
								}
							}
						} else {
							try {
								sleep(100L);
							} catch (InterruptedException exception) {
								/**/log.error(exception.getMessage(), exception);
							}
						}
					} else if (line.equalsIgnoreCase("do?")) {
						//log.debug("response: " + line);
						writeln("mm?");
						//log.debug("mm?");
						/** receive message */
						String service = readLine(100);
						//log.debug("to: " + service);
						if (getContainer().getService(service) != null) {
							writeln("ok.");
							String sender = readLine(100);
							//log.debug("from: " + sender);
							String id = readLine(100);
							//log.debug("message: " + id);
							try {
								Object object = read();
								//log.debug("object: " + object);
								Message message = ((Loader)getContainer().getLoader()).getMessage(id, object);
								getContainer().sendMessage(message, sender, service);
								/**/log.debug("<" + message.getClass().getCanonicalName() + " from " + sender + " to " + service + " :" + message);
								//log.debug("ok.");
								writeln("ok.");
							} catch (StreamCorruptedException exception) {
								/**/log.warn(exception.getMessage());
								writeln("no.");
								//log.debug("no.");
							} catch (ClassNotFoundException exception) {
								/**/log.warn(exception.getMessage());
								writeln("no.");
								//log.debug("no.");
							}
						} else {
							//log.debug("reject: " + service);
							writeln("no.");
						}
					} else {
						try {
							sleep(100L);
						} catch (InterruptedException exception) {
							/**/log.error(exception.getMessage());
						}
					}
				} else { /** not running */
					try {
						sleep(100L);
					} catch (InterruptedException exception) {
						/**/log.error(exception.getMessage());
					}
				}
			}
		} catch (IOException exception) {
			//StackTraceElement[] list  = exception.getStackTrace();
			//String line = exception.getMessage();
			//for (int i = 0; i < list.length; i++) {
			//	line += "\n" + list[i].toString();
			//}
			if (exception instanceof java.net.SocketException) {
				/**/log.info(exception.getMessage());
			} else {
				/**/log.error(exception, exception);
			}
			stoping();
			/** scheduled in order to NOT be transferred to other server */
			getContainer().sendMessage(new ReconnectMessage(getName()), getName(), Cluster.class.getName(), 0);
		} catch (RuntimeException exception) {
			//StackTraceElement[] list = exception.getStackTrace();
			//String line = exception.getMessage();
			//for (int i = 0; i < list.length; i++) {
			//	line += "\n" + list[i].toString();
			//}
			/**/log.error(exception, exception);
			stoping();
		} catch (ServerException exception) {
			/**/log.info(exception.getMessage());
		}
	}

	/**
	 * @return
	 * @throws	IOException
	 * @throws	ClassNotFoundException
	 */
	private Object read() throws IOException, ClassNotFoundException {
		ObjectInputStream in = new ObjectInputStream(conn.getInputStream());
		Object object = in.readObject();
		//log.debug("<<" + object);
		return object;
	}

	/**
	 * Reads a sinle line from the stream.
	 * 
	 * @param	maxCount	max chars into the line
	 * @return	read line from from the stream
	 * @throws	IOException
	 */
	private String readLine(int maxCount) throws IOException {
		String line = "";
		int sign = -1;
		while ((sign = conn.getInputStream().read()) >= 0) {
			//log.gebug(Integer.toHexString(sign) + " ");
			if (sign == '\n') {
				break;
			} else if (sign != '\r') {
				line = line.concat(String.valueOf((char) sign));
				maxCount--;
				if (maxCount < 0) {
					break;
				}
			}
		}
		//log.debug("<<" + line);
		return line;
	}

	/**
	 * @param	object
	 * @throws	IOException
	 */
	private void write(Object object) throws IOException {
		ObjectOutputStream out = new ObjectOutputStream(conn.getOutputStream());
		out.writeObject(object);
		//log.debug(">>" + object);
	}

	private void writeln() throws IOException {
		writeln(null);
	}

	private void writeln(String line) throws IOException {
		if (line != null) {
			//log.debug(">>" + line);
			conn.getOutputStream().write(line.getBytes());
		} else {
			//log.debug(">>");
		}
		conn.getOutputStream().write('\n');
	}
}
